//package com.imooc.bilibili.service.config;
//
//import com.alibaba.fastjson.JSONArray;
//import com.alibaba.fastjson.JSONObject;
//import com.imooc.bilibili.domain.UserFollowing;
//import com.imooc.bilibili.domain.UserMoments;
//import com.imooc.bilibili.domain.constant.UserMomentConstant;
//import com.imooc.bilibili.service.UserFollowingService;
//import com.imooc.bilibili.service.websocket.WebSocketService;
//import io.netty.util.internal.StringUtil;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.core.RedisTemplate;
//
//import java.io.IOException;
//import java.util.ArrayList;
//import java.util.List;
//
////生产者发布消息，消费者获得消息，并把消息存放在redis中
////用户发布动态，那么关注该用户的其他用户都能从redis里得到关于该用户的发布动态提醒
//
///**
// * RocketMQ配置类
// * */
//@Configuration
//public class RocketMQConfig {
//
//    @Value("${rocketmq.name-server}")
//    private String nameServerAddress;
//
//    @Autowired
//    private RedisTemplate<String, String> redisTemplate;
//
//    @Autowired
//    private UserFollowingService userFollowingService;
//
//
//    public static final String MOMENTS_PRODUCER = "momentsProducer";
//    public static final String MOMENTS_CONSUMER = "momentsConsumer";
//    public static final String DANMUS_PRODUCER= "danmusProducer";
//    public static final String DANMUS_CONSUMER = "danmusConsumer";
//
//    /**
//     * 返回生产者实例
//     */
//    @Bean(MOMENTS_PRODUCER)
//    public DefaultMQProducer momentsProducer() throws Exception {
//        //设置生产者所在的分组为GROUP_MOMENTS
//        DefaultMQProducer producer = new DefaultMQProducer(UserMomentConstant.GROUP_MOMENTS);
//        //设置nameServer的IP地址
//        producer.setNamesrvAddr(nameServerAddress);
//        //启动生产者
//        producer.start();
//        //返回生产者
//        return producer;
//    }
//
//
//    /**
//     * 返回消费者
//     */
//    @Bean(MOMENTS_CONSUMER)
//    public DefaultMQPushConsumer momentsConsumer() throws Exception {
//        /*
//        * 1.创建一个消费者实例，并设置分组GROUP_MOMENTS
//        * 2.设置nameServer的IP地址
//        * 3.设置消费者订阅的主题为TOPIC_MOMENTS
//        * 4.注册消息监听器（生产者一旦发布消息，MQ就会把消息推送给消费者，同时做一些额外的操作）
//        * 5.获取生产者发出的消息msg，其实就是接收生产者发出的UserMoments对象（用户动态对象的数据）
//        *   假设我关注的一个up主发布了一条动态，动态信息被发送到了MQ，我可以把这条信息连同其他我关注的up主发布的动态放在一个列表里
//        * */
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentConstant.GROUP_MOMENTS);
//        consumer.setNamesrvAddr(nameServerAddress);
//        consumer.subscribe(UserMomentConstant.TOPIC_MOMENTS, "*");
//        consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
//            MessageExt msg = msgList.get(0);
//            if (msg == null) {
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//            String bodyStr = new String(msg.getBody());
//            UserMoments userMoments = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), UserMoments.class);
//            Long userId = userMoments.getUserId(); //获取发布者userId（方便查找这个userId的所有粉丝）
//            List<UserFollowing> fanList = userFollowingService.getUserFans(userId);//获取发布者的粉丝列表
//
//            /*
//            * 遍历粉丝列表，通知每个粉丝：你们关注的这个up主发布新动态了。
//            * 思路：
//            * 1.通过for循环生成每个粉丝的专属key。例如"subscribed-6"表示为"订阅生产者的用户userId为6"，换句话说就是"userId为6的粉丝"
//            * 2.获取这个粉丝在redis里对应的value，value在redis里是列表形式，因为一个用户可能关注了多个up主，可能会有来自多个up主的动态提醒）
//            * 3.在该 value 的基础上追加 userMoments，再存入redis中
//            * */
//            for (UserFollowing fan : fanList) {
//                String key = "subscribed-" + fan.getUserId();
//                String subscribedListStr = redisTemplate.opsForValue().get(key);
//                List<UserMoments> subscribedList = new ArrayList<>();
//                if (!StringUtil.isNullOrEmpty(subscribedListStr)) {
//                    subscribedList = JSONArray.parseArray(subscribedListStr, UserMoments.class);
//                }
//                subscribedList.add(userMoments);
//                redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));
//            }
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//        consumer.start(); //启动消费者
//        return consumer; //返回消费者对象
//    }
//
//
//    @Bean(DANMUS_PRODUCER)
//    public DefaultMQProducer danmusProducer() throws Exception {
//        //设置生产者所在的分组为GROUP_MOMENTS
//        DefaultMQProducer producer = new DefaultMQProducer(UserMomentConstant.GROUP_DANMUS);
//        //设置nameServer的IP地址
//        producer.setNamesrvAddr(nameServerAddress);
//        //启动生产者
//        producer.start();
//        //返回生产者
//        return producer;
//    }
//
//
//    @Bean(DANMUS_CONSUMER)
//    public DefaultMQPushConsumer danmusConsumer() throws Exception {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentConstant.GROUP_DANMUS);
//        consumer.setNamesrvAddr(nameServerAddress);
//        consumer.subscribe(UserMomentConstant.TOPIC_DANMUS, "*");
//        consumer.registerMessageListener((MessageListenerConcurrently) (msgList, context) -> {
//            MessageExt msg = msgList.get(0);
//            String bodyStr = new String(msg.getBody());
//            JSONObject jb = JSONObject.parseObject(bodyStr);
//
//            //获取消息内的sessionId和message
//            String sessionId = jb.getString("sessionId"); //客户端和服务端的会话id
//            String message = jb.getString("message"); //弹幕的内容
//
//            WebSocketService webSocketService = WebSocketService.WEBSOCKET_MAP.get(sessionId);
//            if (webSocketService.getSession().isOpen()) {
//                try {
//                    webSocketService.sendMessage(message);
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//            //标记该消息已经被成功消费
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
//        //启动消费者
//        consumer.start();
//        //返回消费者对象
//        return consumer;
//    }
//}
